"""
Case Type   : 权限
Case Name   : show grants--显示访问域的权限
Create At   : 2024.8.19
Owner       : wang-tianjing1
Description :
    1、创建用户
    2、创建域
    3、授予权限
    4、清理环境
Expect      :
    1、成功
    2、成功
    3、成功
    4、成功
History     :
    modified by @xuwenfei1 2025-3.19:在执行show grants for user前，将参数b_compatibility_user_host_auth的值设置为true
"""

import os
import unittest

from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger
from yat.test import Node
from yat.test import macro


class OpreatorTest(unittest.TestCase):

    def setUp(self):
        self.log = Logger()
        self.log.info(f'-----{os.path.basename(__file__)} start-----')
        self.common = Common()
        self.sh_primary = CommonSH('PrimaryDbBmode')
        self.user_node = Node('PrimaryDbBmode')
        self.constant = Constant()
        self.db_user = 'u_sg_tester08'
        self.db_domain = 'd_domain_0008'
        status = self.sh_primary.get_db_cluster_status()
        self.assertTrue("Degraded" in status or "Normal" in status)

    def test_show(self):
        text = '----step1:创建用户;expect:成功----'
        self.log.info(text)
        sql_cmd = f'set current_schema to public;' \
                  f'show current_schema;' \
                  f"drop user if exists {self.db_user} cascade;" \
                  f"create user {self.db_user} password" \
                  f"'{macro.COMMON_PASSWD}';" \
                  f"grant all privileges to {self.db_user};"
        self.log.info(sql_cmd)
        sql_res = self.sh_primary.execut_db_sql(sql_cmd)
        self.log.info(sql_res)
        self.assertIn('SET', sql_res, '执行失败' + text)
        self.assertIn('public', sql_res, '执行失败' + text)
        self.assertIn('DROP ROLE', sql_res, '执行失败' + text)
        self.assertIn('CREATE ROLE', sql_res, '执行失败' + text)
        self.assertIn('ALTER ROLE', sql_res, '执行失败' + text)

        text = '----step2:创建域;expect:成功----'
        self.log.info(text)
        self.log.info("===修改support_extended_features = 'on'，except：重启设置成功===")
        res = self.sh_primary.execute_gsguc('set',
                                            self.constant.GSGUC_SUCCESS_MSG,
                                            "support_extended_features = 'on'")
        self.log.info(res)
        self.assertTrue(res, '执行失败:' + text)
        restart = self.sh_primary.restart_db_cluster()
        self.assertTrue(restart, '执行失败:' + text)
        self.log.info("===创建域，except：成功===")
        sql_cmd2 = f'drop domain if exists {self.db_domain} cascade;' \
                   f'create domain {self.db_domain} as int check(value > 0);' \
                   f'set b_compatibility_user_host_auth  = true;' \
                   f'show grants for \'{self.db_user}\';' \
                   f'select * from pg_type where typname = \'{self.db_domain}\';' \
                   f'show current_schema;'
        self.log.info(sql_cmd2)
        connect = f'-U {self.db_user} -W {macro.COMMON_PASSWD}'
        sql_res2 = self.sh_primary.execut_db_sql(sql_cmd2, sql_type=connect)
        self.log.info(sql_res2)
        self.assertIn('DROP DOMAIN', sql_res2, '执行失败' + text)
        self.assertIn('CREATE DOMAIN', sql_res2, '执行失败' + text)
        self.assertIn('SET', sql_res2, '执行失败' + text)
        self.assertIn('ALTER USER u_sg_tester08 WITH LOGIN SYSADMIN', sql_res2, '执行失败' + text)
        self.assertIn('d_domain_0008', sql_res2, '执行失败' + text)
        self.assertIn('"$user",public', sql_res2, '执行失败' + text)

        text = '----step3:授予权限,expect:成功----'
        self.log.info(text)
        sql_cmd3 = f'grant usage on domain {self.db_domain} to {self.db_user};' \
                   f'set b_compatibility_user_host_auth  = true;' \
                   f'show grants for \'{self.db_user}\';'
        self.log.info(sql_cmd3)
        sql_res3 = self.sh_primary.execut_db_sql(sql_cmd3, sql_type=connect)
        self.log.info(sql_res3)
        self.assertIn('GRANT', sql_res3, '执行失败' + text)
        self.assertIn('SET', sql_res3, '执行失败' + text)
        self.assertIn('(2 rows)', sql_res3, '执行失败' + text)

        sql_cmd4 = f'grant usage on domain {self.db_domain} to {self.db_user} with grant option;' \
                   f'set b_compatibility_user_host_auth  = true;' \
                   f'show grants for \'{self.db_user}\';' \
                   f'select * from pg_type where typname = \'{self.db_domain}\';'
        self.log.info(sql_cmd4)
        sql_res4 = self.sh_primary.execut_db_sql(sql_cmd4, sql_type=connect)
        self.log.info(sql_res4)
        self.assertIn('SET', sql_res4, '执行失败' + text)
        self.assertIn('GRANT', sql_res4, '执行失败' + text)
        self.assertIn('(2 rows)', sql_res4, '执行失败' + text)
        self.assertIn('d_domain_0008', sql_res4, '执行失败' + text)

        sql_cmd5 = f'revoke usage on domain {self.db_domain} from {self.db_user};' \
                   f'set b_compatibility_user_host_auth  = true;' \
                   f'show grants for \'{self.db_user}\';' \
                   f'select * from pg_type where typname = \'{self.db_domain}\';'
        self.log.info(sql_cmd5)
        sql_res5 = self.sh_primary.execut_db_sql(sql_cmd5, sql_type=connect)
        self.log.info(sql_res5)
        self.assertIn('REVOKE', sql_res5, '执行失败' + text)
        self.assertIn('SET', sql_res5, '执行失败' + text)
        self.assertIn('(1 row)', sql_res5, '执行失败' + text)
        self.assertIn('d_domain_0008', sql_res5, '执行失败' + text)

        sql_cmd6 = f'grant all on domain {self.db_domain} to {self.db_user};' \
                   f'set b_compatibility_user_host_auth  = true;' \
                   f'show grants for \'{self.db_user}\';'
        self.log.info(sql_cmd6)
        sql_res6 = self.sh_primary.execut_db_sql(sql_cmd6, sql_type=connect)
        self.log.info(sql_res6)
        self.assertIn('SET', sql_res6, '执行失败' + text)
        self.assertIn('GRANT', sql_res6, '执行失败' + text)
        self.assertIn('(3 rows)', sql_res6, '执行失败' + text)

        sql_cmd7 = f'grant all on domain {self.db_domain} to {self.db_user} with grant option;' \
                   f'set b_compatibility_user_host_auth  = true;' \
                   f'show grants for \'{self.db_user}\';' \
                   f'select * from pg_type where typname = \'{self.db_domain}\';'
        self.log.info(sql_cmd7)
        sql_res7 = self.sh_primary.execut_db_sql(sql_cmd7, sql_type=connect)
        self.log.info(sql_res7)
        self.assertIn('SET', sql_res7, '执行失败' + text)
        self.assertIn('GRANT', sql_res7, '执行失败' + text)
        self.assertIn('(3 rows)', sql_res7, '执行失败' + text)
        self.assertIn('d_domain_0008', sql_res7, '执行失败' + text)

        sql_cmd8 = f'revoke all on domain {self.db_domain} from {self.db_user};' \
                   f'set b_compatibility_user_host_auth  = true;' \
                   f'show grants for \'{self.db_user}\';' \
                   f'select * from pg_type where typname = \'{self.db_domain}\';'
        self.log.info(sql_cmd8)
        sql_res8 = self.sh_primary.execut_db_sql(sql_cmd8, sql_type=connect)
        self.log.info(sql_res8)
        self.assertIn('REVOKE', sql_res8, '执行失败' + text)
        self.assertIn('SET', sql_res8, '执行失败' + text)
        self.assertIn('(1 row)', sql_res8, '执行失败' + text)
        self.assertIn('d_domain_0008', sql_res8, '执行失败' + text)

        sql_cmd9 = f'grant usage on domain {self.db_domain} to {self.db_user};' \
                   f'set b_compatibility_user_host_auth  = true;' \
                   f'show grants for \'{self.db_user}\';'
        self.log.info(sql_cmd9)
        sql_res9 = self.sh_primary.execut_db_sql(sql_cmd9, sql_type=connect)
        self.log.info(sql_res9)
        self.assertIn('GRANT', sql_res9, '执行失败' + text)
        self.assertIn('(2 rows)', sql_res9, '执行失败' + text)

        sql_cmd10 = f'grant all privileges on domain {self.db_domain} to {self.db_user} with grant option;' \
                    f'set b_compatibility_user_host_auth  = true;' \
                    f'show grants for \'{self.db_user}\';' \
                    f'select * from pg_type where typname = \'{self.db_domain}\';'
        self.log.info(sql_cmd10)
        sql_res10 = self.sh_primary.execut_db_sql(sql_cmd10, sql_type=connect)
        self.log.info(sql_res10)
        self.assertIn('GRANT', sql_res10, '执行失败' + text)
        self.assertIn('(3 rows)', sql_res10, '执行失败' + text)
        self.assertIn('d_domain_0008', sql_res10, '执行失败' + text)

        sql_cmd11 = f'revoke all privileges on domain {self.db_domain} from {self.db_user};' \
                    f'set b_compatibility_user_host_auth  = true;' \
                    f'show grants for \'{self.db_user}\';' \
                    f'select * from pg_type where typname = \'{self.db_domain}\';'
        self.log.info(sql_cmd11)
        sql_res11 = self.sh_primary.execut_db_sql(sql_cmd11, sql_type=connect)
        self.log.info(sql_res11)
        self.assertIn('REVOKE', sql_res11, '执行失败' + text)
        self.assertIn('SET', sql_res11, '执行失败' + text)
        self.assertIn('(1 row)', sql_res11, '执行失败' + text)
        self.assertIn('d_domain_0008', sql_res11, '执行失败' + text)

    def tearDown(self):
        text = '--step4:清理环境;expect:成功--'
        self.log.info(text)
        sql_cmd = f'drop domain if exists {self.db_domain};' \
                  f'set current_schema to public;'
        self.log.info(sql_cmd)
        connect = f'-U {self.db_user} -W {macro.COMMON_PASSWD}'
        sql_res = self.sh_primary.execut_db_sql(sql_cmd, sql_type=connect)
        self.log.info(sql_res)
        sql_clean = f'drop owned by {self.db_user} cascade;' \
                    f"drop user {self.db_user};" \
                    f"drop domain if exists {self.db_domain} cascade;"
        self.log.info(sql_clean)
        clean_res = self.sh_primary.execut_db_sql(sql_clean)
        self.log.info(clean_res)
        self.assertNotIn("ERROR", sql_res, '执行失败' + text)
        self.log.info("===恢复support_extended_features，except：重启设置成功===")
        res = self.sh_primary.execute_gsguc('set',
                                            self.constant.GSGUC_SUCCESS_MSG,
                                            "support_extended_features")
        self.log.info(res)
        self.assertTrue(res, '执行失败:' + text)
        restart = self.sh_primary.restart_db_cluster()
        self.assertTrue(restart, '执行失败:' + text)
        self.log.info(f'-----{os.path.basename(__file__)} end-----')
